package io.kiki.sba.registry.server.shared.remoting;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.api.Client;
import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.api.exchange.NodeExchanger;
import io.kiki.sba.registry.api.exchange.RequestChannelClosedException;
import io.kiki.sba.registry.api.exchange.RequestException;
import io.kiki.sba.registry.api.exchange.message.Request;
import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.util.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;


public abstract class ServerSideExchanger implements NodeExchanger {

    @Autowired
    protected DataExchanger boltDataExchanger;

    @Override
    public Response request(Request request) throws RequestException {
        final Url url = request.getRequestUrl();
        if (url == null) {
            throw new RequestException("null url", request);
        }
        return request(url, request);
    }

    public Response request(Url url, Request request) throws RequestException {
        final Server server = boltDataExchanger.getServer(getServerPort());
        if (server == null) {
            throw new RequestException("no server for " + url + "," + getServerPort(), request);
        }
        final int timeout = request.getTimeout() != null ? request.getTimeout() : getRpcTimeoutMillis();
        Channel channel = null;
        if (url == null) {
            // now use in dsr console sync case
            channel = chooseChannel(server);
        } else {
            channel = server.getChannel(url);
        }

        if (channel == null || !channel.isConnected()) {
            throw new RequestChannelClosedException(getServerPort() + ", channel may be closed, " + url, request);
        }
        if (request.getCallBack() != null) {
            server.sendCallback(channel, request.getRequestBody(), request.getCallBack(), timeout);
            return () -> Response.ResultStatus.SUCCESSFUL;
        } else {
            final Object result = server.sendSync(channel, request.getRequestBody(), timeout);
            return () -> result;
        }
    }

    private Channel chooseChannel(Server server) {
        List<Channel> channels = server.getChannelList();
        return CollectionUtils.getRandom(channels);
    }

    @Override
    public Client connectServer() {
        throw new UnsupportedOperationException();
    }

    public abstract int getRpcTimeoutMillis();

    public abstract int getServerPort();
}
